package com.sen.gulimall.order.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * @author Kisen
 * @email lqs0911@qq.com
 * @date 2022/11/18 11:59
 * @detail
 */
@Slf4j
@Configuration
public class MQSenderUtil {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 依赖注入rabbitTemplate之后再设置它的回调对象
     * Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
     */
    @PostConstruct
    public void init() {

        /**
         * 交换机不管是否收到消息的一个回调方法
         * 1.CorrelationData 消息相关数据
         * 2.ack 交换机是否收到消息
         * 3.cause 未收到消息的原因
         */
        RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
            String id = correlationData != null ? correlationData.getId() : "";
            if (ack) {
                log.info("交换机已经接收到id为：{}的消息", id);

            } else {
                //交换机未接收到的消息，需要重试发送消息
                log.error("交换机未接收到id为：{}的消息，由于原因：{}", id, cause);

            }
        };

        /**
         * 消息回退：只有当消息传递过程中不可达目的地时才将消息返回给生产者
         */
        RabbitTemplate.ReturnsCallback returnsCallback = returned -> {
            //被退回的消息保存到备份交换机，由备份交换机来进行转发和处理
            String errorMsg = String.format("消息：%s被交换机退回，退回原因：%s，交换机是：%s，路由key：%s",
                    new String(returned.getMessage().getBody()), returned.getReplyText(), returned.getExchange(), returned.getRoutingKey());
            log.error(errorMsg);

            String messageId = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            log.error("被退回的消息ID：{}", messageId);

        };

        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnsCallback(returnsCallback);
    }

    /**
     * 发送消息(ERP)
     *
     * @param exchange   交换机名
     * @param routingKey 路由key名
     * @param message    消息
     */
    public void sendMessage(String exchange, String routingKey, Object message) {
        CorrelationData correlationData = new CorrelationData(String.valueOf(SnowflakeIdGenerator.generateUniqueId()));
        log.info("生产者发送消息id:{}, message:{}, exchange:{}, routingKey:{}", correlationData.getId(), message, exchange, routingKey);

        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor -> {
                //设置消息持久化
                messagePostProcessor.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return messagePostProcessor;
            }, correlationData);

        } catch (Exception e) {
            log.error("生产者发送消息失败id:{}, message:{}, exchange:{}, routingKey:{}, 异常原因：{}", correlationData.getId(), message, exchange, routingKey, e);
        }

    }

    /**
     * 消息重试
     *
     * @param exchange
     * @param routingKey
     * @param messageId
     * @param message
     */
    public void retryMessage(String exchange, String routingKey, String messageId, Object message) {
        CorrelationData correlationData = new CorrelationData(messageId);
        log.info("ERP生产者重试消息id:{}, message:{}, exchange:{}, routingKey:{}", messageId, message, exchange, routingKey);

        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor -> {
                //设置消息持久化
                messagePostProcessor.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return messagePostProcessor;
            }, correlationData);

        } catch (Exception e) {
            log.error("ERP生产者重试消息失败id:{}, message:{}, exchange:{}, routingKey:{}, 异常原因：{}", correlationData.getId(), message, exchange, routingKey, e);
        }
    }
}
